package com.foreveross.fetl.queue;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.inject.Inject;

import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.apache.tools.zip.ZipOutputStream;
import org.springframework.stereotype.Component;

import com.common.remote.ftp.FtpConfig;
import com.common.remote.ftp.FtpTransferException;
import com.common.remote.sftp.SFtpFileTransferClient;
import com.foreveross.fetl.application.ExcelInfoApplication;
import com.foreveross.fetl.common.cfg.system.ReportProperties;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.common.vo.hangxin.PlaneInfo;
import com.foreveross.fetl.common.vo.model.doub.DoublePlaneInfo;
import com.foreveross.fetl.core.domain.task.SourceEntity;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.JsonUtil;
import com.foreveross.fetl.util.PlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;
import com.foreveross.fetl.util.ZipUtils;
import com.foreveross.fetl.vo.ExcelInfoVO;

@Component
public class MetadataDoublePreprocess {
	private static Logger log = Logger.getLogger("MetadataDouble");
	
	@Inject
	private ExcelInfoApplication excelInfoApplication;
	
	private int tasksNumber;//任务数 多少条rowkey
	
	private int failuresNumber;//失败数
	
	private int insertNumber; //插入条数 
	
	private long excuteStartTime;
	

	
	private String taskTime=null;
	
	public void  rollBackNumber(){
		tasksNumber=0;
		failuresNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("往返元数据生成", this.getClass().getSimpleName());
	}
	public void  rollBackAllNumber(){
		tasksNumber=0;
		failuresNumber=0;
		insertNumber=0;
		excuteStartTime = System.currentTimeMillis();
		LogMessageUtils.create("当天往返元数据生成", this.getClass().getSimpleName());
	}
	
	public void execute(){
		rollBackNumber();
		Calendar calendar=Calendar.getInstance();
		calendar.add(Calendar.HOUR_OF_DAY, -1);
		process(calendar.getTime());
		LogMessageUtils.build("往返元数据生成", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
	}
	
	
	public void executeAll(){
		rollBackAllNumber();
		Calendar calendar=Calendar.getInstance();
		calendar.add(Calendar.DAY_OF_MONTH, -1);
		processAll(calendar.getTime()); 
		LogMessageUtils.build("当天往返元数据生成", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
	}
	
	
	public void manualExecute(String start){
		rollBackAllNumber();
		Date date =DateUtils.parse(start, "yyyyMMdd");
		processAll(date);
		LogMessageUtils.build("当天往返元数据生成", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
	}
	

	public void manualExecute(final String start,String time){
		this.rollBackNumber();
		try {
		List<String> timeList = Arrays.asList(time.split(";"));
		taskTime = start + " " +time;
		for (String s : timeList) {
			String str =start;
			str += s;
			Date startTime = DateUtils.parse(str, "yyyyMMddHH");
			this.process(startTime);
		}
		LogMessageUtils.build("往返元数据生成", this.getClass().getSimpleName(), 2, null, 
				failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime);
		} catch (Exception e) {
			log.error(StringUtils.getStackTrace(e));
			LogMessageUtils.build("往返元数据生成", this.getClass().getSimpleName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber, 
					insertNumber, taskTime, excuteStartTime);
			
		}
	}
	private void process(Date time) {
		SourceEntity sourceEntity = new SourceEntity();
		List<SourceEntity> listSource =sourceEntity.findAll();
		Map<String ,String> mapFile=new HashMap<String,String>(32);
		//重新初始化汇率
		ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));
		File files =null;
		for(SourceEntity source:listSource){
			tasksNumber++;
			this.findFormHbase(source,time); // 创建文件
			files=new File(ReportProperties.getFilePath()+"double/newFile/"+source.getHbaseKey());
		    File[] list =files.listFiles();
			if(null!=list && list.length>0){
				FileOutputStream fous=null;
				ZipOutputStream zipOut =null;
				File uploadFile=null;
				File file =null;
				try {
					uploadFile= new File(ReportProperties.getFilePath()+"double/"+source.getHbaseKey()+"_"+DateUtils.format(time, "yyyy-MM-dd_HH")+".zip");
					if (!uploadFile.exists()) {
						uploadFile.createNewFile(); 
					}
					fous=new FileOutputStream(uploadFile);
					zipOut =new ZipOutputStream(fous);
					 for (int i=0; i<list.length; i++) {
			               if (list[i].isFile()) {
			            	   if(list[i].getName().indexOf(DateUtils.format(time, "yyyy-MM-dd_HH"))<=0){
			            		  continue;
			            	   }
			            	   file=list[i];
			            	   ZipUtils.zipFile(file,zipOut); 
			            	   file.delete(); 
			               }
			              
				       }
					zipOut.flush();
					zipOut.close();
					zipOut=null;
					fous.flush();
					fous.close();
				}catch (Exception e) {
					failuresNumber++;
					log.error(StringUtils.getStackTrace(e));
					FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
							this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e);
					FetlErrorUtils.saveToHbase(message);
				}finally {
					if(null!=file && file.exists()){
						file.delete();
					}
				
					if(null!=zipOut){
						try {
							zipOut.flush();
							zipOut.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
					if(null!=fous){
						try {
							fous.flush();
							fous.close();
						} catch (IOException e) {
							e.printStackTrace();
						}
					}
				}
				mapFile.put(source.getHbaseKey(), uploadFile.getPath());
			}
		}
		File allFile = new File(ReportProperties.getFilePath()+"double/"+DateUtils.format(time, "yyyy-MM-dd_HH")+".zip");
		FileOutputStream fousAll = null;
		ZipOutputStream zipOutAll = null;
		File fileDir=null;
		boolean filebind =false;
		try {
			fousAll=new FileOutputStream(allFile); 
			zipOutAll=new ZipOutputStream(fousAll);
			for(String hbaseKey:mapFile.keySet()){
					
					File f =new File(mapFile.get(hbaseKey));
					try {
						//zipFile(f,zipOutAll);
						filebind=true;
				        //上传
				        try {
							this.uploadFile(f, hbaseKey, time,1);
						} catch (Exception e) {
							try {
								this.uploadFile(f, hbaseKey, time,1);
							} catch (Exception e1) {
								log.error(StringUtils.getStackTrace(e));
							}
						}
				      //  f.delete(); //先不删除，每天打包再删除
				       
				       //解压文件
				       ZipUtils.unzip(f.getPath(), ReportProperties.getFilePath()+"double/"+hbaseKey);
				       fileDir=new File( ReportProperties.getFilePath()+"double/"+hbaseKey);
				       File[] list =fileDir.listFiles();
				       for (int i=0; i<list.length; i++) {
			               if (list[i].isFile()) {
			            	   ZipUtils.zipFile(list[i],zipOutAll); 
			               }
				       }
				       ZipUtils.delAllFile(fileDir.getPath());
					} catch (Exception e) {
						failuresNumber++;
						log.error(StringUtils.getStackTrace(e));
						FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
								this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e);
						FetlErrorUtils.saveToHbase(message);
					}finally {
						if(f.exists()){
							f.delete();
						}
						if(fileDir.listFiles().length>0){
						  ZipUtils.delAllFile(fileDir.getPath());
						}
					}
				}
			if(filebind){
				zipOutAll.flush();
				zipOutAll.close();
				zipOutAll=null;
				fousAll.flush();
				fousAll.close();
				 try {
					 	this.uploadFile(allFile, "all", time,1);
					} catch (Exception e) {
						try {
							this.uploadFile(allFile, "all", time,1);
						} catch (Exception e1) {
							log.error(StringUtils.getStackTrace(e));
							String str=DateUtils.format(time, "yyyyMMddHH");
							FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
									this.getClass().getSimpleName(),str, e);
							FetlErrorUtils.saveToHbase(message);
						}
					}
				
				allFile.delete();
			}
		} catch (Exception e) {
			failuresNumber++;
			log.error(StringUtils.getStackTrace(e));
			FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
					this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e);
			FetlErrorUtils.saveToHbase(message);
		}finally {
			if(allFile.exists()){
				allFile.delete();
			}
			if(null!=zipOutAll){
				try {
					zipOutAll.flush();
					zipOutAll.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if(null!=fousAll){
				try {
					fousAll.flush();
					fousAll.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		
	}
	
	private void findFormHbase(SourceEntity source, Date time) {
		List<PlaneInfo> list =new ArrayList<PlaneInfo>(1024);
		String startRow=RowkeyUtils.getRowKey(time, source.getHbaseKey(),  RowkeyUtils.START_ROW_KEY);
		String endRow=RowkeyUtils.getRowKey(time, source.getHbaseKey(),  RowkeyUtils.STOP_ROW_KEY);
		ResultScanner rs = null;
		int thousand=0;
		String path=ReportProperties.getFilePath()+"double/newFile/"+source.getHbaseKey()+"/";
		List<String> listkeys=new ArrayList<String>();
		try {
			rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
					startRow, endRow);
			for (Result res : rs) {
				try {// 某一条航班数据的错误，不应该影响其他航班数据的预处理
					Map<String, Map<String, String>> curMap = HBaseHelper
							.transformHBaseRowToMap(res);
					Map<String, String> curPalneInfoMap = curMap.get("planeinfo");
					String id =new String(res.getRow());
					listkeys.add(id);
//					float flightPrice = Float.parseFloat(curPalneInfoMap.get("lowerPrice") != null 
//							 ? curPalneInfoMap.get("lowerPrice") : "0");
//					if (flightPrice <= 0) {
//						log.info("rowkey[" + new String(res.getRow())+ "]  flightPrice="+flightPrice);
//						continue;
//					}
					DoublePlaneInfo planeInfo= null;

					double refePrice=ExchangeRateUtil.getRefePrice(curPalneInfoMap.get("currency"));
					//TODO
					if(curPalneInfoMap.get("grabChannelNum").equals("00006") && null==curPalneInfoMap.get("currency")){
						refePrice=ExchangeRateUtil.getRefePrice("USD");
					}
					planeInfo=HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, refePrice,DoublePlaneInfo.class);
					if(null!=planeInfo && null!=planeInfo.getFlightNo()){
						List<PlaneInfo> list2=PlaneInfoUtil.changeToPlaneInfo(planeInfo);
						if(null!=list2 && list2.size()>0){
							for(PlaneInfo vo:list2){
								list.add(vo);
							}
						}
					}
					
					if(null!=list && list.size()>=500){
						String json=JsonUtil.getJSONString(list);
						list.clear();
						thousand++;
						String fileName=source.getHbaseKey()+"_"+DateUtils.format(time, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
						ZipUtils.creatFile(path, fileName, json);
						listkeys.clear();
					}
				}catch(Exception e){
					log.error(StringUtils.getStackTrace(e));
					FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
							this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e);
					message.setKeys(listkeys);
					FetlErrorUtils.saveToHbase(message);
					listkeys.clear();
				}
			}
			
			if(null!=list && list.size()>0){
				String json=JsonUtil.getJSONString(list);
				list.clear();
				thousand++;
				String fileName=source.getHbaseKey()+"_"+DateUtils.format(time, "yyyy-MM-dd_HH")+"_"+thousand+".txt";
				ZipUtils.creatFile(path, fileName, json);
				listkeys.clear();
			}
		} catch (Exception e) {
			log.error(StringUtils.getStackTrace(e));
			FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataDouble,
					this.getClass().getSimpleName(),DateUtils.format(time, "yyyyMMddHH"), e);
			message.setKeys(listkeys);
			FetlErrorUtils.saveToHbase(message);
			listkeys.clear();
		}finally{
			if(rs != null) rs.close();
			rs = null;
		}
		
	}
	
	private void processAll(Date date) {
		SourceEntity sourceEntity = new SourceEntity();
		List<SourceEntity> listSource =sourceEntity.findAll();
		List<ExcelInfoVO> listExcelInfo=null;
		List<String> listCode =new ArrayList<String>();
		for(SourceEntity source:listSource){
			listCode.add(source.getHbaseKey());
		}
		listCode.add("all");
		for(String source:listCode){
			tasksNumber++;
			listExcelInfo=excelInfoApplication.findBycode(source, DateUtils.format(date, "yyyyMMdd"),"all",2);
			if(null==listExcelInfo || listExcelInfo.isEmpty()){
				continue;
			}
			FileOutputStream fous=null;
			ZipOutputStream zipOut =null;
			File downloadFile=null;
			File file=null;
			File fileDir=null;
			try {
				if("all".equals(source)){
					downloadFile= new File(DateUtils.format(date, "yyyy-MM-dd")+".zip");
				}else{
					downloadFile= new File(source+"_"+DateUtils.format(date, "yyyy-MM-dd")+".zip");
				}
				if (!downloadFile.exists()) {
					downloadFile.createNewFile(); 
				}
				fous=new FileOutputStream(downloadFile);
				zipOut =new ZipOutputStream(fous);
				for(ExcelInfoVO vo :listExcelInfo){
					try {
						file =new File(ReportProperties.getFilePath()+"double/"+vo.getName());
						if(vo.getName().equals(downloadFile.getName())){
							continue;
						}
						//System.out.println(vo.getName());
						if(null==file || !file.exists()) {
							file=this.downloadFile(vo);//如果本地没有就从服务器下载 
						}
						if(null == file || !file.exists()){
							continue;
						}
						
						   ZipUtils.unzip(file.getPath(), ReportProperties.getFilePath()+"double/"+source);
					       fileDir=new File( ReportProperties.getFilePath()+"double/"+source);
					       File[] list =fileDir.listFiles();
					       for (int i=0; i<list.length; i++) {
				               if (list[i].isFile()) {
				            	   ZipUtils.zipFile(list[i],zipOut); 
				               }
					       }
					       ZipUtils.delAllFile(fileDir.getPath());
						file.delete();
					} catch (Exception e) {
						e.printStackTrace();
						FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataAllDouble,
								this.getClass().getSimpleName(),DateUtils.format(date, "yyyyMMddHH"), e);
						FetlErrorUtils.saveToHbase(message);
					}
				}
				zipOut.flush();
				zipOut.close();
				zipOut=null;
				fous.flush();
				fous.close();
				
				 try {
					 	this.uploadFile(downloadFile, source, date,2);
					} catch (Exception e) {
						try {
							this.uploadFile(downloadFile, source, date,2);
						} catch (Exception e1) {
							log.error(StringUtils.getStackTrace(e));
							FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataAllDouble,
									this.getClass().getSimpleName(),DateUtils.format(date, "yyyyMMddHH"), e);
							FetlErrorUtils.saveToHbase(message);
						}
					}
				 
				downloadFile.delete();
			} catch (Exception e) {
				failuresNumber++;
				e.printStackTrace();
				FetlErrorMessage message = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataAllDouble,
						this.getClass().getSimpleName(),DateUtils.format(date, "yyyyMMddHH"), e);
				FetlErrorUtils.saveToHbase(message);
			}finally{
				try {
					delLocaleFile();
					if(null!=file){
						file.delete();
					}
					if(null!=downloadFile){
						downloadFile.delete();
					}
					if(null!=zipOut){
						zipOut.flush();
						zipOut.close();
					}
					if(null!=fous){
						fous.flush();
						fous.close();
					}
					if(fileDir.listFiles().length>0){
						  ZipUtils.delAllFile(fileDir.getPath());
						}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}
		
	}
	
	
	private File downloadFile(ExcelInfoVO vo) {
		String address =ReportProperties.getAddress();
		int port= Integer.parseInt(null==ReportProperties.getPort()? "22":ReportProperties.getPort());
		String userName= ReportProperties.getUserName();
		String passWord =ReportProperties.getPassword();
		String path=ReportProperties.getPath();
		if(!path.endsWith("/")) path+="/";
		InputStream is = null;
		File file =new File(vo.getName());
		FtpConfig config =null;
		SFtpFileTransferClient sftpClient=null;
		boolean flag = false;
		try {
			config=new FtpConfig(address,port,userName,passWord);
			sftpClient=new SFtpFileTransferClient(config);
			try {
				flag = sftpClient.connect();
			} catch (FtpTransferException e) {
			}
			if (!flag) {
				log.error("connect sftp fail");
				return null;
			}
			String src=path+vo.getPath();
			if(!sftpClient.isFileExists(src)){
				return null;
			}
			is =sftpClient.downloadStream(src);
			ZipUtils.inputstreamtofile(is, file);
			sftpClient.disconnect();
		} catch (FtpTransferException e) {
			e.printStackTrace();
		}finally{
			if(null!=is){
				try {
					is.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if(null!=sftpClient){
				sftpClient.disconnect();
			}
		}
		
		return file;
	}
	
	private void delLocaleFile(){
		List<ExcelInfoVO> listExcelInfo=null;
		try {
			listExcelInfo = excelInfoApplication.findByTime(DateUtils.format(
					new Date(), "yyyyMMdd") + "00","all",2);
			File file = null;
			if (null != listExcelInfo && listExcelInfo.size() > 0) {
				for (ExcelInfoVO vo : listExcelInfo) {
					file = new File(ReportProperties.getFilePath()+"double/"
							+ vo.getName());
					if (null != file && file.exists()) {
						file.delete();
					}
				}
			}
		} catch (Exception e) {
			log.error("delete local file error");
		}
	}
	
	
	
	
	
	private void uploadFile(File f, String hbasekey, Date time,int type) {
		String address =ReportProperties.getAddress();
		int port= Integer.parseInt(null==ReportProperties.getPort()? "22":ReportProperties.getPort());
		String userName= ReportProperties.getUserName();
		String passWord =ReportProperties.getPassword();
		String path=ReportProperties.getPath();
		if(!path.endsWith("/")) path+="/";
		String dateType =DateUtils.format(time, "yyyyMMddHH");
		if(type==2){
			dateType=DateUtils.format(time, "yyyyMMdd");
		}
		String dir=path +"round-trip/"+dateType.substring(0, 4)+"/"+dateType.substring(4, 6)+"/"+dateType.substring(6, 8)+"/"+hbasekey;
		
		//ftp连接配置
		FtpConfig config =new FtpConfig(address,port,userName,passWord);
		SFtpFileTransferClient sftpClient=new SFtpFileTransferClient(config);
		InputStream in =null;
		boolean flag=false;
		try {
			try {
				flag = sftpClient.connect();
			} catch (Exception e) {
				log.error(StringUtils.getStackTrace(e));
				Thread.sleep(20 * 1000L);
				flag = sftpClient.connect();
			}
			if(flag){
				//如果目录不存在则创建
				String[] dirs=dir.split("/");
				StringBuffer buffer=new StringBuffer();
				for(String d:dirs){
					buffer.append(d+"/");
					if(!sftpClient.isDirectoryExists(buffer.toString())){
						sftpClient.makeDirectory(buffer.toString());
					}
				}
				in=new FileInputStream(f);
				sftpClient.upload(in, dir+"/"+f.getName());
				sftpClient.disconnect();
				insertNumber++;
				ExcelInfoVO oldVo =excelInfoApplication.findByName(f.getName(),"all",2);
				if(null!=oldVo){
					excelInfoApplication.delete(oldVo);
				}
				ExcelInfoVO excelInfoVO =new ExcelInfoVO();
				excelInfoVO.setCode(hbasekey);
				excelInfoVO.setDateType(dateType);
				if(type==2){
					excelInfoVO.setDateType(dateType+"00");
				}
				excelInfoVO.setFileType("all");
				excelInfoVO.setType(2);
				excelInfoVO.setName(f.getName());
				excelInfoVO.setPath("round-trip/"+dateType.substring(0, 4)+"/"+dateType.substring(4, 6)+"/"+dateType.substring(6, 8)+"/"+hbasekey+"/"+f.getName());
				excelInfoVO.setUploadDate(new Date());
				excelInfoApplication.saveOrUpdate(excelInfoVO);
				in.close();
			}
		} catch (Exception e) {
			try {
				throw e;
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		}finally{
			if(null!=sftpClient){
				sftpClient.disconnect();	
			}
			try {
			if (in != null) {
				in.close();
			}
		} catch (Exception e) {
		}
		}
	}
}
